package org.az.netty_study.p16_rpc;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.util.concurrent.DefaultPromise;
import lombok.extern.slf4j.Slf4j;
import org.az.netty_study.p14_netty综合案例2_聊天室.protocol.MessageCodecSharable;
import org.az.netty_study.p14_netty综合案例2_聊天室.protocol.ProcotolFrameDecoder;
import org.az.netty_study.p16_rpc.handler.RpcResponseMessageHandler;
import org.az.netty_study.p16_rpc.message.RpcReqMessage;
import org.az.netty_study.p16_rpc.service.HelloService;

import java.lang.reflect.Proxy;

/**
 * 什么是rpc
 * 远程过程调用，remote process call，运行在两个服务器上的两个不同的应用A和B，A想要使用B应用提供的方法/函数
 * 由于A和B不在一个内存空间，无法直接调用，需要A将方法/函数的名称、返回值类型、参数类型，参数值告诉B，然后B拿着这些数据找到目标方法/函数并调用，将方法/函数返回值传给A
 * 这个过程涉及网络通信、协议、序列化反序列化
 *
 * 下面利用netty+动态代理实现了一个简单的rpc框架
 */
@Slf4j
public class RpcClient {

    private static Channel channel = null;


    public static void main(String[] args) {
        HelloService helloService = getProxyService(HelloService.class);
        String hello = helloService.hello("yt");
        System.out.println(hello);
    }

    static <T> T getProxyService(Class<T> clazz) {
        //创建动态代理，将rpc代码放在代理类中
        Object o = Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{clazz}, ((proxy, method, args) -> {
            if (channel == null) {
                synchronized (RpcClient.class) {
                    if (channel == null) {
                        initChannel();
                    }
                }
            }

            RpcReqMessage message = new RpcReqMessage(
                    clazz.getName(),
                    method.getName(),
                    method.getReturnType(),
                    method.getParameterTypes(),
                    args
            );
            message.setSequenceId(SequenceIdGenerator.getSequenceId());
            //发送消息
            channel.writeAndFlush(message);

            //准备Promise，用来接收结果
            //如果Promise的泛型指定为？，则只能取不能放
            DefaultPromise<Object> promise = new DefaultPromise<>(channel.eventLoop());
            RpcResponseMessageHandler.map.put(message.getSequenceId(), promise);
            //直到Promise里面有东西了，才继续运行，否则阻塞
            promise.await();
           // 也可以通过异步方式获取值
//            promise.addListener(p -> {
//            });


            //返回结果
            if (promise.isSuccess()) {
                return promise.getNow();
            } else {
                Throwable cause = promise.cause();
                throw new RuntimeException(cause);
            }
        }));


        return (T) o;
    }

    private static void initChannel() {
        NioEventLoopGroup group = new NioEventLoopGroup();
        LoggingHandler LOGGING_HANDLER = new LoggingHandler(LogLevel.DEBUG);
        MessageCodecSharable MESSAGE_CODEC = new MessageCodecSharable();

        // rpc 响应消息处理器，待实现
        RpcResponseMessageHandler RPC_HANDLER = new RpcResponseMessageHandler();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.group(group);
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(new ProcotolFrameDecoder());
                    ch.pipeline().addLast(LOGGING_HANDLER);
                    ch.pipeline().addLast(MESSAGE_CODEC);
                    ch.pipeline().addLast(RPC_HANDLER);
                    ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                        @Override
                        public void channelActive(ChannelHandlerContext ctx) throws Exception {
//                            ctx.writeAndFlush(message).addListener(promise -> {
//                                if (!promise.isSuccess()) {
//                                    Throwable cause = promise.cause();
//                                    log.error("error", cause);
//                                }
//                            });
                        }
                    });
                }
            });
            channel = bootstrap.connect("localhost", 8080).sync().channel();

            channel.closeFuture().addListener(c -> {
                group.shutdownGracefully();
            });
        } catch (Exception e) {
            log.error("client error", e);
        }
    }
}
